/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef DLL_EXPORT
#  define USE_STATIC_LIB
#endif

#if defined(__CYGWIN__)
#define USE_IPV6
#endif

#include <zookeeper.h>
#include <zookeeper.jute.h>
#include <proto.h>
#include <block_storage.h>
#include "zk_adaptor.h"
#include "zookeeper_log.h"
#include "zk_hashtable.h"

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <poll.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>
#include <stdarg.h>
#include <limits.h>

#include "config.h"

#ifdef HAVE_SYS_UTSNAME_H
#include <sys/utsname.h>
#endif

#ifdef HAVE_GETPWUID_R
#include <pwd.h>
#endif

#define IF_DEBUG(x) if(logLevel==ZOO_LOG_LEVEL_DEBUG) {x;}

const int ZOOKEEPER_WRITE = 1 << 0;
const int ZOOKEEPER_READ = 1 << 1;

const int ZOO_EPHEMERAL = 1 << 0;
const int ZOO_SEQUENCE = 1 << 1;

const int ZOO_EXPIRED_SESSION_STATE = EXPIRED_SESSION_STATE_DEF;
const int ZOO_AUTH_FAILED_STATE = AUTH_FAILED_STATE_DEF;
const int ZOO_CONNECTING_STATE = CONNECTING_STATE_DEF;
const int ZOO_ASSOCIATING_STATE = ASSOCIATING_STATE_DEF;
const int ZOO_CONNECTED_STATE = CONNECTED_STATE_DEF;
static __attribute__ ((unused)) const char* state2String(int state){
    switch(state){
    case 0:
        return "ZOO_CLOSED_STATE";
    case CONNECTING_STATE_DEF:
        return "ZOO_CONNECTING_STATE";
    case ASSOCIATING_STATE_DEF:
        return "ZOO_ASSOCIATING_STATE";
    case CONNECTED_STATE_DEF:
        return "ZOO_CONNECTED_STATE";
    case EXPIRED_SESSION_STATE_DEF:
        return "ZOO_EXPIRED_SESSION_STATE";
    case AUTH_FAILED_STATE_DEF:
        return "ZOO_AUTH_FAILED_STATE";
    }
    return "INVALID_STATE";
}

const int ZOO_CREATED_EVENT = CREATED_EVENT_DEF;
const int ZOO_DELETED_EVENT = DELETED_EVENT_DEF;
const int ZOO_CHANGED_EVENT = CHANGED_EVENT_DEF;
const int ZOO_CHILD_EVENT = CHILD_EVENT_DEF;
const int ZOO_SESSION_EVENT = SESSION_EVENT_DEF;
const int ZOO_NOTWATCHING_EVENT = NOTWATCHING_EVENT_DEF;
static __attribute__ ((unused)) const char* watcherEvent2String(int ev){
    switch(ev){
    case 0:
        return "ZOO_ERROR_EVENT";
    case CREATED_EVENT_DEF:
        return "ZOO_CREATED_EVENT";
    case DELETED_EVENT_DEF:
        return "ZOO_DELETED_EVENT";
    case CHANGED_EVENT_DEF:
        return "ZOO_CHANGED_EVENT";
    case CHILD_EVENT_DEF:
        return "ZOO_CHILD_EVENT";
    case SESSION_EVENT_DEF:
        return "ZOO_SESSION_EVENT";
    case NOTWATCHING_EVENT_DEF:
        return "ZOO_NOTWATCHING_EVENT";
    }
    return "INVALID_EVENT";
}

const int ZOO_PERM_READ = 1 << 0;
const int ZOO_PERM_WRITE = 1 << 1;
const int ZOO_PERM_CREATE = 1 << 2;
const int ZOO_PERM_DELETE = 1 << 3;
const int ZOO_PERM_ADMIN = 1 << 4;
const int ZOO_PERM_ALL = 0x1f;

#define COMPLETION_WATCH -1
#define COMPLETION_VOID 0
#define COMPLETION_STAT 1
#define COMPLETION_DATA 2
#define COMPLETION_STRINGLIST 3
#define COMPLETION_STRINGLIST_STAT 4
#define COMPLETION_ACLLIST 5
#define COMPLETION_STRING 6



const char*err2string(int err);
static int queue_session_event(zhandle_t *zh, int state);
static const char* format_endpoint_info(const struct sockaddr_storage* ep);
static const char* format_current_endpoint_info(zhandle_t* zh, int sid);

static int handle_socket_error_msg(zhandle_t *zh, int line, int rc, int sid,
    const char* format,...);
static void cleanup_bufs(zhandle_t *zh,int callCompletion,int rc, int sid);

static int disable_conn_permute=0; // permute enabled by default


static void *SYNCHRONOUS_MARKER = (void*)&SYNCHRONOUS_MARKER;
static int isValidPath(const char* path, const int flags);

const void *zoo_get_context(zhandle_t *zh)
{
    return zh->context;
}

void zoo_set_context(zhandle_t *zh, void *context)
{
    if (zh != NULL) {
        zh->context = context;
    }
}

int zoo_recv_timeout(zhandle_t *zh)
{
    return zh->recv_timeout;
}


int is_unrecoverable(zhandle_t *zh)
{
    return (zh->state<0)? ZINVALIDSTATE: ZOK;
}


/**
 * Frees and closes everything associated with a handle,
 * including the handle itself.
 */
static void destroy(zhandle_t *zh)
{
    int i;
    printf("destroy zookeeper\n");
    if (zh == NULL) {
        return;
    }
    /* call any outstanding completions with a special error code */
    for(i=0; i<SERVER_COUNT; i++)
    	cleanup_bufs(zh,1,ZCLOSING, i);
    if (zh->hostname != 0) {
        free(zh->hostname);
        zh->hostname = NULL;
    }
    for(i=0; i<SERVER_COUNT; i++){
    	if (zh->fd[i] != -1) {
            close(zh->fd[i]);
            zh->fd[i] = -1;
            zh->state[i] = 0;
    	}
    }
    if (zh->addrs != 0) {
        free(zh->addrs);
        zh->addrs = NULL;
    }


}

static void setup_random()
{
    int seed;
    int fd = open("/dev/urandom", O_RDONLY);
    if (fd == -1) {
        seed = getpid();
    } else {
        int rc = read(fd, &seed, sizeof(seed));
        assert(rc == sizeof(seed));
        close(fd);
    }
    srandom(seed);
}

#ifndef __CYGWIN__
/**
 * get the errno from the return code 
 * of get addrinfo. Errno is not set
 * with the call to getaddrinfo, so thats
 * why we have to do this.
 */
static int getaddrinfo_errno(int rc) { 
    switch(rc) {
    case EAI_NONAME:
    case EAI_NODATA:
        return ENOENT;
    case EAI_MEMORY:
        return ENOMEM;
    default:
        return EINVAL;
    }
}
#endif

/**
 * fill in the addrs array of the zookeeper servers in the zhandle. after filling
 * them in, we will permute them for load balancing.
 */
int getaddrs(zhandle_t *zh)
{
    char *hosts = strdup(zh->hostname);
    char *host;
    char *strtok_last;
    struct sockaddr_storage *addr;
    int i;
    int rc;
    int alen = 0; /* the allocated length of the addrs array */

    zh->addrs_count = 0;
    if (zh->addrs) {
        free(zh->addrs);
        zh->addrs = 0;
    }
    if (!hosts) {
         LOG_ERROR(("out of memory"));
        errno=ENOMEM;
        return ZSYSTEMERROR;
    }
    zh->addrs = 0;
    host=strtok_r(hosts, ",", &strtok_last);
    while(host) {
        char *port_spec = strrchr(host, ':');
        char *end_port_spec;
        int port;
        if (!port_spec) {
            LOG_ERROR(("no port in %s", host));
            errno=EINVAL;
            rc=ZBADARGUMENTS;
            goto fail;
        }
        *port_spec = '\0';
        port_spec++;
        port = strtol(port_spec, &end_port_spec, 0);
        if (!*port_spec || *end_port_spec) {
            LOG_ERROR(("invalid port in %s", host));
            errno=EINVAL;
            rc=ZBADARGUMENTS;
            goto fail;
        }
#if defined(__CYGWIN__)
        // sadly CYGWIN doesn't have getaddrinfo
        // but happily gethostbyname is threadsafe in windows
        {
        struct hostent *he;
        char **ptr;
        struct sockaddr_in *addr4;

        he = gethostbyname(host);
        if (!he) {
            LOG_ERROR(("could not resolve %s", host));
            errno=ENOENT;
            rc=ZBADARGUMENTS;
            goto fail;
        }

        /* Setup the address array */
        for(ptr = he->h_addr_list;*ptr != 0; ptr++) {
            if (zh->addrs_count == alen) {
                alen += 16;
                zh->addrs = realloc(zh->addrs, sizeof(*zh->addrs)*alen);
                if (zh->addrs == 0) {
                    LOG_ERROR(("out of memory"));
                    errno=ENOMEM;
                    rc=ZSYSTEMERROR;
                    goto fail;
                }
            }
            addr = &zh->addrs[zh->addrs_count];
            addr4 = (struct sockaddr_in*)addr;
            addr->ss_family = he->h_addrtype;
            if (addr->ss_family == AF_INET) {
                addr4->sin_port = htons(port);
                memset(&addr4->sin_zero, 0, sizeof(addr4->sin_zero));
                memcpy(&addr4->sin_addr, *ptr, he->h_length);
                zh->addrs_count++;
            }
#if defined(AF_INET6)
            else if (addr->ss_family == AF_INET6) {
                struct sockaddr_in6 *addr6;

                addr6 = (struct sockaddr_in6*)addr;
                addr6->sin6_port = htons(port);
                addr6->sin6_scope_id = 0;
                addr6->sin6_flowinfo = 0;
                memcpy(&addr6->sin6_addr, *ptr, he->h_length);
                zh->addrs_count++;
            }
#endif
            else {
                LOG_WARN(("skipping unknown address family %x for %s",
                         addr->ss_family, zh->hostname));
            }
        }
        host = strtok_r(0, ",", &strtok_last);
        }
#else
        {
        struct addrinfo hints, *res, *res0;

        memset(&hints, 0, sizeof(hints));
#ifdef AI_ADDRCONFIG
        hints.ai_flags = AI_ADDRCONFIG;
#else
        hints.ai_flags = 0;
#endif
        hints.ai_family = AF_UNSPEC;
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_protocol = IPPROTO_TCP;

        while(isspace(*host) && host != strtok_last)
            host++;

        if ((rc = getaddrinfo(host, port_spec, &hints, &res0)) != 0) {
            //bug in getaddrinfo implementation when it returns
            //EAI_BADFLAGS or EAI_ADDRFAMILY with AF_UNSPEC and 
            // ai_flags as AI_ADDRCONFIG
            if ((hints.ai_flags == AI_ADDRCONFIG) && 
                ((rc ==EAI_BADFLAGS) || (rc == EAI_ADDRFAMILY))) {
                //reset ai_flags to null
                hints.ai_flags = 0;
                //retry getaddrinfo
                rc = getaddrinfo(host, port_spec, &hints, &res0);
            }
            if (rc != 0) {
                errno = getaddrinfo_errno(rc);
                LOG_ERROR(("getaddrinfo: %s\n", strerror(errno)));
                rc=ZSYSTEMERROR;
                goto fail;
            }
        }

        for (res = res0; res; res = res->ai_next) {
            // Expand address list if needed
            if (zh->addrs_count == alen) {
                void *tmpaddr;
                alen += 16;
                tmpaddr = realloc(zh->addrs, sizeof(*zh->addrs)*alen);
                if (tmpaddr == 0) {
                    LOG_ERROR(("out of memory"));
                    errno=ENOMEM;
                    rc=ZSYSTEMERROR;
                    goto fail;
                }
                zh->addrs=tmpaddr;
            }

            // Copy addrinfo into address list
            addr = &zh->addrs[zh->addrs_count];
            switch (res->ai_family) {
            case AF_INET:
#if defined(AF_INET6)
            case AF_INET6:
#endif
                memcpy(addr, res->ai_addr, res->ai_addrlen);
                ++zh->addrs_count;
                break;
            default:
                LOG_WARN(("skipping unknown address family %x for %s",
                res->ai_family, zh->hostname));
                break;
            }
        }

        freeaddrinfo(res0);

        host = strtok_r(0, ",", &strtok_last);
        }
#endif
    }
    free(hosts);

    /*if(!disable_conn_permute){
        setup_random();
        for(i = 0; i < zh->addrs_count; i++) {
            struct sockaddr_storage *s1 = zh->addrs + random()%zh->addrs_count;
            struct sockaddr_storage *s2 = zh->addrs + random()%zh->addrs_count;
            if (s1 != s2) {
                struct sockaddr_storage t = *s1;
                *s1 = *s2;
                *s2 = t;
            }
        }
    }*/
    return ZOK;
fail:
    if (zh->addrs) {
        free(zh->addrs);
        zh->addrs=0;
    }
    if (hosts) {
        free(hosts);
    }
    return rc;
}

const clientid_t *zoo_client_id(zhandle_t *zh)
{
    return &zh->client_id;
}



static void log_env() {
  char buf[2048];
#ifdef HAVE_SYS_UTSNAME_H
  struct utsname utsname;
#endif

#if defined(HAVE_GETUID) && defined(HAVE_GETPWUID_R)
  struct passwd pw;
  struct passwd *pwp = NULL;
  uid_t uid = 0;
#endif

  LOG_INFO(("Client environment:zookeeper.version=%s", PACKAGE_STRING));

#ifdef HAVE_GETHOSTNAME
  gethostname(buf, sizeof(buf));
  LOG_INFO(("Client environment:host.name=%s", buf));
#else
  LOG_INFO(("Client environment:host.name=<not implemented>"));
#endif

#ifdef HAVE_SYS_UTSNAME_H
  uname(&utsname);
  LOG_INFO(("Client environment:os.name=%s", utsname.sysname));
  LOG_INFO(("Client environment:os.arch=%s", utsname.release));
  LOG_INFO(("Client environment:os.version=%s", utsname.version));
#else
  LOG_INFO(("Client environment:os.name=<not implemented>"));
  LOG_INFO(("Client environment:os.arch=<not implemented>"));
  LOG_INFO(("Client environment:os.version=<not implemented>"));
#endif

#ifdef HAVE_GETLOGIN
  LOG_INFO(("Client environment:user.name=%s", getlogin()));
#else
  LOG_INFO(("Client environment:user.name=<not implemented>"));
#endif

#if defined(HAVE_GETUID) && defined(HAVE_GETPWUID_R)
  uid = getuid();
  if (!getpwuid_r(uid, &pw, buf, sizeof(buf), &pwp)) {
    LOG_INFO(("Client environment:user.home=%s", pw.pw_dir));
  } else {
    LOG_INFO(("Client environment:user.home=<NA>"));
  }
#else
  LOG_INFO(("Client environment:user.home=<not implemented>"));
#endif

#ifdef HAVE_GETCWD
  if (!getcwd(buf, sizeof(buf))) {
    LOG_INFO(("Client environment:user.dir=<toolong>"));
  } else {
    LOG_INFO(("Client environment:user.dir=%s", buf));
  }
#else
  LOG_INFO(("Client environment:user.dir=<not implemented>"));
#endif
}

/**
 * Create a zookeeper handle associated with the given host and port.
 */
zhandle_t *zookeeper_init(const char *host, 
  int recv_timeout, const clientid_t *clientid, void *context, int flags, void *storage)
{
    int errnosave = 0;
    int i=0;
    zhandle_t *zh = NULL;

    log_env();
    LOG_INFO(("Initiating client connection, host=%s sessionTimeout=%d "
          " sessionId=%#llx sessionPasswd=%s context=%p flags=%d",
              host,
              recv_timeout,
              (clientid == 0 ? 0 : clientid->client_id),
              ((clientid == 0) || (clientid->passwd == 0) ?
               "<null>" : "<hidden>"),
              context,
              flags));

    zh = calloc(1, sizeof(*zh));
    zh -> storage = storage;
    if (!zh) {
        return 0;
    }
    for(i=0; i<SERVER_COUNT; i++){
    	zh->fd[i] = -1;
    	zh->state[i] = 0;
    }
    zh->context = context;
    zh->recv_timeout = recv_timeout;
    zh->hostname = strdup(host);
    if (host == 0 || *host == 0) { // what we shouldn't dup
        errno=EINVAL;
        goto abort;
    }
    if (zh->hostname == 0) {
        goto abort;
    }
    if(getaddrs(zh)!=0) {
        goto abort;
    }
    if (clientid) {
        memcpy(&zh->client_id, clientid, sizeof(zh->client_id));
    } else {
        memset(&zh->client_id, 0, sizeof(zh->client_id));
    }
    for(i=0; i<SERVER_COUNT; i++){
        zh->primer_buffer[i].buffer = zh->primer_storage_buffer[i];
        zh->primer_buffer[i].curr_offset = 0;
    	zh->primer_buffer[i].len = sizeof(zh->primer_storage_buffer[i]);
    	zh->primer_buffer[i].next = 0;

    	zh->next_deadline[i].tv_sec=zh->next_deadline[i].tv_usec=0;
    	zh->socket_readable[i].tv_sec=zh->socket_readable[i].tv_usec=0;
    }
    zh->last_zxid = 0;

    if (adaptor_init(zh) == -1) {
        goto abort;
    }

    return zh;
abort:
    errnosave=errno;
    destroy(zh);
    free(zh);
    errno=errnosave;
    return 0;
}

void get_live_servers(zhandle_t *zh, int *nodes){
    int i;
    for(i=0; i<SERVER_COUNT; i++){
	if(zh->fd[i]!=-1 && zh->state[i] == ZOO_CONNECTED_STATE)
	    nodes[i]=1;
	else
	    nodes[i]=0;
    }
}

int get_num_live_servers(zhandle_t *zh){
    int i;
    int ret = 0;
    for(i=0; i<SERVER_COUNT; i++){
        if(zh->fd[i]!=-1 && zh->state[i] == ZOO_CONNECTED_STATE)
            ret++;
    }
    return ret;
}


static buffer_list_t *allocate_buffer(char *buff, int len)
{
    buffer_list_t *buffer = calloc(1, sizeof(*buffer));
    if (buffer == 0)
        return 0;

    buffer->len = len==0?sizeof(*buffer):len;
    buffer->curr_offset = 0;
    buffer->buffer = buff;
    buffer->next = 0;
    return buffer;
}

static void free_buffer(buffer_list_t *b)
{
    if (!b) {
        return;
    }
    if (b->buffer) {
        free(b->buffer);
    }
    free(b);
}

static buffer_list_t *dequeue_buffer(buffer_head_t *list)
{
    buffer_list_t *b;
    lock_buffer_list(list);
    b = list->head;
    if (b) {
        list->head = b->next;
        if (!list->head) {
            assert(b == list->last);
            list->last = 0;
        }
    }
    unlock_buffer_list(list);
    return b;
}

static int remove_buffer(buffer_head_t *list)
{
    buffer_list_t *b = dequeue_buffer(list);
    if (!b) {
        return 0;
    }
    free_buffer(b);
    return 1;
}

static void queue_buffer(buffer_head_t *list, buffer_list_t *b, int add_to_front)
{
    b->next = 0;
    lock_buffer_list(list);
    if (list->head) {
        assert(list->last);
        // The list is not empty
        if (add_to_front) {
            b->next = list->head;
            list->head = b;
        } else {
            list->last->next = b;
            list->last = b;
        }
    }else{
        // The list is empty
        assert(!list->head);
        list->head = b;
        list->last = b;
    }
    unlock_buffer_list(list);
}

static int queue_buffer_bytes(buffer_head_t *list, char *buff, int len)
{
    buffer_list_t *b  = allocate_buffer(buff,len);
    if (!b)
        return ZSYSTEMERROR;
    queue_buffer(list, b, 0);
    return ZOK;
}

int send_bytes_to_server(zhandle_t *zh, char *buff, int len, int sid){
    //printf("send_bytes_to_server %d\n", sid);
    enter_critical(zh);
    queue_buffer_bytes(&zh->to_send[sid], buff, len);
    leave_critical(zh);

    adaptor_send_queue(zh, 0, sid);
    return 0;
}


static int queue_front_buffer_bytes(buffer_head_t *list, char *buff, int len)
{
    buffer_list_t *b  = allocate_buffer(buff,len);
    if (!b)
        return ZSYSTEMERROR;
    queue_buffer(list, b, 1);
    return ZOK;
}

static __attribute__ ((unused)) int get_queue_len(buffer_head_t *list)
{
    int i;
    buffer_list_t *ptr;
    lock_buffer_list(list);
    ptr = list->head;
    for (i=0; ptr!=0; ptr=ptr->next, i++)
        ;
    unlock_buffer_list(list);
    return i;
}
/* returns:
 * -1 if send failed,
 * 0 if send would block while sending the buffer (or a send was incomplete),
 * 1 if success
 */
static int send_buffer(int fd, buffer_list_t *buff)
{
    int len = buff->len;
    int off = buff->curr_offset;
    int rc = -1;

    if (off < 4) {
        /* we need to send the length at the beginning */
        int nlen = htonl(len);
        char *b = (char*)&nlen;
        rc = send(fd, b + off, sizeof(nlen) - off, MSG_NOSIGNAL);
        if (rc == -1) {
            if (errno != EAGAIN) {
                return -1;
            } else {
                return 0;
            }
        } else {
            buff->curr_offset  += rc;
	    //printf("send len %d %d\n", len, rc);
        }
        off = buff->curr_offset;
    }
    if (off >= 4) {
        /* want off to now represent the offset into the buffer */
        off -= sizeof(buff->len);
        rc = send(fd, buff->buffer + off, len - off, MSG_NOSIGNAL);
        if (rc == -1) {
            if (errno != EAGAIN) {
                return -1;
            }
        } else {
            buff->curr_offset += rc;
        }
    }
    return buff->curr_offset == len + sizeof(buff->len);
}

/* returns:
 * -1 if recv call failed,
 * 0 if recv would block,
 * 1 if success
 */
static int recv_buffer(int fd, buffer_list_t *buff)
{
    int off = buff->curr_offset;
    int rc = 0;
    //fprintf(LOGSTREAM, "rc = %d, off = %d, line %d\n", rc, off, __LINE__);

    /* if buffer is less than 4, we are reading in the length */
    if (off < 4) {
        char *buffer = (char*)&(buff->len);
        rc = recv(fd, buffer+off, sizeof(int)-off, 0);
        //printf("rc = %d, off = %d, line %d\n", rc, off, __LINE__);
        switch(rc) {
        case 0:
            errno = EHOSTDOWN;
        case -1:
            if (errno == EAGAIN) {
                return 0;
            }
            return -1;
        default:
            buff->curr_offset += rc;
        }
        off = buff->curr_offset;
        if (buff->curr_offset == sizeof(buff->len)) {
            buff->len = ntohl(buff->len);
            buff->buffer = calloc(1, buff->len);
        }
    }
    if (buff->buffer) {
        /* want off to now represent the offset into the buffer */
        off -= sizeof(buff->len);

        rc = recv(fd, buff->buffer+off, buff->len-off, 0);
        //printf("rc = %d, off = %d, line %d\n", rc, off, __LINE__);
        switch(rc) {
        case 0:
            errno = EHOSTDOWN;
        case -1:
            if (errno == EAGAIN) {
                break;
            }
            return -1;
        default:
            buff->curr_offset += rc;
        }
    }
    return buff->curr_offset == buff->len + sizeof(buff->len);
}

void free_buffers(buffer_head_t *list)
{
    while (remove_buffer(list))
        ;
}


static void cleanup_bufs(zhandle_t *zh,int callCompletion,int rc, int sid)
{
    enter_critical(zh);
    free_buffers(&zh->to_send[sid]);
    free_buffers(&zh->to_process[sid]);
    leave_critical(zh);
    if (zh->input_buffer[sid] && zh->input_buffer[sid] != &zh->primer_buffer[sid]) {
        free_buffer(zh->input_buffer[sid]);
        zh->input_buffer[sid] = 0;
    }
}

static void handle_error(zhandle_t *zh,int rc)
{
    int i = 0;
    for(i=0; i<SERVER_COUNT; i++){
    	close(zh->fd[i]);
    	if (is_unrecoverable(zh)) {
            LOG_DEBUG(("Calling a watcher for a ZOO_SESSION_EVENT and the state=%s",
                state2String(zh->state[i])));
    	} else if (zh->state[i] == ZOO_CONNECTED_STATE) {
            LOG_DEBUG(("Calling a watcher for a ZOO_SESSION_EVENT and the state=CONNECTING_STATE"));
    	}
    	cleanup_bufs(zh,1,rc, i);
    	zh->fd[i] = -1;
    }
    if (!is_unrecoverable(zh)) {
	for(i=0; i<SERVER_COUNT; i++)
            zh->state[i] = 0;
    }
}

static int handle_socket_error_msg(zhandle_t *zh, int line, int rc, int sid,
        const char* format, ...)
{
    if(logLevel>=ZOO_LOG_LEVEL_ERROR){
        va_list va;
        char buf[1024];
        va_start(va,format);
        vsnprintf(buf, sizeof(buf)-1,format,va);
        log_message(ZOO_LOG_LEVEL_ERROR,line,__func__,
            format_log_message("Socket [%s] zk retcode=%d, errno=%d(%s): %s",
            format_current_endpoint_info(zh, sid),rc,errno,strerror(errno),buf));
        va_end(va);
    }
    cleanup_bufs(zh,1,rc, sid);
    close(zh->fd[sid]);
    zh->fd[sid] = -1;
    //handle_error(zh,rc);
    return rc;
}


static void free_key_list(char **list, int count)
{
    int i;

    for(i = 0; i < count; i++) {
        free(list[i]);
    }
    free(list);
}


static int serialize_prime_connect(struct connect_req *req, char* buffer){
    //this should be the order of serialization
    int offset = 0;

    req->lastZxidSeen = htonll(req->lastZxidSeen);
    memcpy(buffer + offset, &req->lastZxidSeen, sizeof(req->lastZxidSeen));
    offset = offset +  sizeof(req->lastZxidSeen);

    req->timeOut = htonl(req->timeOut);
    memcpy(buffer + offset, &req->timeOut, sizeof(req->timeOut));
    offset = offset +  sizeof(req->timeOut);

    req->sessionId = htonll(req->sessionId);
    memcpy(buffer + offset, &req->sessionId, sizeof(req->sessionId));
    offset = offset +  sizeof(req->sessionId);

    return 0;
}

static int deserialize_prime_response(struct prime_struct *req, char* buffer){
     //this should be the order of deserialization
     int offset = 0;
     memcpy(&req->len, buffer + offset, sizeof(req->len));
     offset = offset +  sizeof(req->len);

     req->len = ntohl(req->len);

     memcpy(&req->timeOut, buffer + offset, sizeof(req->timeOut));
     offset = offset +  sizeof(req->timeOut);

     req->timeOut = ntohl(req->timeOut);
     memcpy(&req->sessionId, buffer + offset, sizeof(req->sessionId));
     offset = offset +  sizeof(req->sessionId);

     req->sessionId = htonll(req->sessionId);
     memcpy(&req->cxid, buffer + offset, sizeof(req->cxid));
     offset = offset +  sizeof(req->cxid);

     req->cxid = ntohl(req->cxid);
     return 0;
 }

static int prime_connection(zhandle_t *zh, int sid)
{
    int rc;
    /*this is the size of buffer to serialize req into*/
    char buffer_req[HANDSHAKE_REQ_SIZE];
    int len = sizeof(buffer_req);
    int hlen = 0;
    struct connect_req req;
    req.sessionId = zh->client_id.client_id;
    req.timeOut = zh->recv_timeout;
    req.lastZxidSeen = zh->last_zxid;
    hlen = htonl(len);
    /* We are running fast and loose here, but this string should fit in the initial buffer! */
    rc=send(zh->fd[sid], &hlen, sizeof(len), 0);
    serialize_prime_connect(&req, buffer_req);
    rc=rc<0 ? rc : send(zh->fd[sid], buffer_req, len, 0);
    if (rc<0) {
        return handle_socket_error_msg(zh, __LINE__, ZCONNECTIONLOSS, sid,
                "failed to send a handshake packet: %s", strerror(errno));
    }
    zh->state[sid] = ZOO_ASSOCIATING_STATE;

    zh->input_buffer[sid] = &zh->primer_buffer[sid];
    /* This seems a bit weird to to set the offset to 4, but we already have a
     * length, so we skip reading the length (and allocating the buffer) by
     * saying that we are already at offset 4 */
    zh->input_buffer[sid]->curr_offset = 4;

    return ZOK;
}

static inline int calculate_interval(const struct timeval *start,
        const struct timeval *end)
{
    int interval;
    struct timeval i = *end;
    i.tv_sec -= start->tv_sec;
    i.tv_usec -= start->tv_usec;
    interval = i.tv_sec * 1000 + (i.tv_usec/1000);
    return interval;
}

static struct timeval get_timeval(int interval)
{
    struct timeval tv;
    if (interval < 0) {
        interval = 0;
    }
    tv.tv_sec = interval/1000;
    tv.tv_usec = (interval%1000)*1000;
    return tv;
}


 int send_ping(zhandle_t* zh, int sid)
 {
    int rc;
    struct oarchive *oa = create_buffer_oarchive();
    struct RequestHeader h = { .xid = PING_XID, .type = PING_OP };

    rc = serialize_RequestHeader(oa, "header", &h);
    enter_critical(zh);
    gettimeofday(&zh->last_ping[sid], 0);
    rc = rc < 0 ? rc : queue_buffer_bytes(&zh->to_send[sid], get_buffer(oa),
            get_buffer_len(oa));
    leave_critical(zh);
    close_buffer_oarchive(&oa, 0);
    return rc<0 ? rc : adaptor_send_queue(zh, 0, sid);
}

 int zookeeper_interest(zhandle_t *zh, int *fd, int *interest,
     struct timeval *tv, int sid)
{
    struct timeval now;
    if(DEBUG_SHARD)
    printf("zookeeper_interest %d = %d\n", sid, zh->fd[sid]);
    if(zh==0 || fd==0 ||interest==0 || tv==0)
        return ZBADARGUMENTS;
    if (is_unrecoverable(zh))
        return ZINVALIDSTATE;
    gettimeofday(&now, 0);
    if(zh->next_deadline[sid].tv_sec!=0 || zh->next_deadline[sid].tv_usec!=0){
        int time_left = calculate_interval(&zh->next_deadline[sid], &now);
        if (time_left > 10)
            LOG_WARN(("Exceeded deadline by %dms", time_left));
    }
    api_prolog(zh);
    *fd = zh->fd[sid];
    *interest = 0;
    tv->tv_sec = 0;
    tv->tv_usec = 0;
    if (*fd == -1) {
        if (1==0) {
        }else {
            int rc;
            int on = 1;
	    struct timeval connect_interval = get_timeval(zh->recv_timeout/3);
	    if(zh->next_connect[sid].tv_sec!=0){
		int idle_connection = calculate_interval(&now, &zh->next_connect[sid]);
		//printf("idle_connection[%d]= %d\n", sid, idle_connection);
		if(idle_connection>0){
                    //zh->next_connect[sid].tv_sec = now.tv_sec + connect_interval.tv_sec;
                    //zh->next_connect[sid].tv_usec = now.tv_usec + connect_interval.tv_usec;
		    *tv = get_timeval(zh->recv_timeout/3);
		    return api_epilog(zh,ZOK);
                }
            }
            zh->fd[sid] = socket(zh->addrs[sid].ss_family, SOCK_STREAM, 0);
	    zh->next_connect[sid].tv_sec = now.tv_sec + connect_interval.tv_sec;
            zh->next_connect[sid].tv_usec = now.tv_usec + connect_interval.tv_usec;
            if (zh->fd[sid] < 0) {
                return api_epilog(zh,handle_socket_error_msg(zh,__LINE__,
                                                             ZSYSTEMERROR, sid, "socket() call failed"));
            }
            setsockopt(zh->fd[sid], IPPROTO_TCP, TCP_NODELAY, &on, sizeof(int));
	    //setsockopt(zh->fd[sid], SOL_SOCKET, MSG_NOSIGNAL, &on, sizeof(int));

            fcntl(zh->fd[sid], F_SETFL, O_NONBLOCK|fcntl(zh->fd[sid], F_GETFL, 0));
#if defined(AF_INET6)
            if (zh->addrs[sid].ss_family == AF_INET6) {
                rc = connect(zh->fd[sid], (struct sockaddr*) &zh->addrs[sid], sizeof(struct sockaddr_in6));
            } else {
#else
            {
#endif
		fprintf(stderr, "%d %d\n", sid, ((struct sockaddr_in*) &zh->addrs[sid])->sin_addr.s_addr);
                rc = connect(zh->fd[sid], (struct sockaddr*) &zh->addrs[sid], sizeof(struct sockaddr_in));
            }
            if (rc == -1) {
                /* we are handling the non-blocking connect according to
                 * the description in section 16.3 "Non-blocking connect"
                 * in UNIX Network Programming vol 1, 3rd edition */
                if (errno == EWOULDBLOCK || errno == EINPROGRESS)
                    zh->state[sid] = ZOO_CONNECTING_STATE;
                else
                    return api_epilog(zh,handle_socket_error_msg(zh,__LINE__,
                            ZCONNECTIONLOSS, sid, "connect() call failed"));
            } else {
                if((rc=prime_connection(zh, sid))!=0)
                    return api_epilog(zh,rc);

                LOG_INFO(("Initiated connection to server [%s]",
                        format_endpoint_info(&zh->addrs[sid])));
            }
        }
        *fd = zh->fd[sid];
        *tv = get_timeval(zh->recv_timeout/3);
        zh->last_recv[sid] = now;
        zh->last_send[sid] = now;
        zh->last_ping[sid] = now;
    }
    if (zh->fd[sid] != -1) {
        /*int idle_recv = calculate_interval(&zh->last_recv[sid], &now);
        int idle_send = calculate_interval(&zh->last_send[sid], &now);
        int recv_to = zh->recv_timeout*2/3 - idle_recv;
        int send_to = zh->recv_timeout/3;
        // have we exceeded the receive timeout threshold?
        if (recv_to <= 0) {
            // We gotta cut our losses and connect to someone else
            errno = ETIMEDOUT;
            *fd=-1;
            *interest=0;
            *tv = get_timeval(0);
            return api_epilog(zh,handle_socket_error_msg(zh,
                    __LINE__,ZOPERATIONTIMEOUT, sid,
                    "connection timed out (exceeded timeout by %dms)",-recv_to));
        }
	//printf("state[%d]=%d\n", sid, zh->state[sid]);
        // We only allow 1/3 of our timeout time to expire before sending
        // a PING
        if (zh->state[sid]==ZOO_CONNECTED_STATE) {
            send_to = zh->recv_timeout/3 - idle_send;
            if (send_to <= 0) {
//                LOG_DEBUG(("Sending PING to %s (exceeded idle by %dms)",
//                                format_current_endpoint_info(zh),-send_to));
		//printf("send ping\n");
                int rc=send_ping(zh, sid);
                if (rc < 0){
                    LOG_ERROR(("failed to send PING request (zk retcode=%d)",rc));
                    return api_epilog(zh,rc);
                }
                send_to = zh->recv_timeout/3;
            }
        }
        // choose the lesser value as the timeout
        *tv = get_timeval(recv_to < send_to? recv_to:send_to);
        zh->next_deadline[sid].tv_sec = now.tv_sec + tv->tv_sec;
        zh->next_deadline[sid].tv_usec = now.tv_usec + tv->tv_usec;
        if (zh->next_deadline[sid].tv_usec > 1000000) {
            zh->next_deadline[sid].tv_sec += zh->next_deadline[sid].tv_usec / 1000000;
            zh->next_deadline[sid].tv_usec = zh->next_deadline[sid].tv_usec % 1000000;
        }*/
	*tv = get_timeval(zh->recv_timeout/3);
        *interest = ZOOKEEPER_READ;
        /* we are interested in a write if we are connected and have something
         * to send, or we are waiting for a connect to finish. */
        if (zh->to_send[sid].head && (zh->state[sid] == ZOO_CONNECTED_STATE)){
            *interest |= ZOOKEEPER_WRITE;
        }
        if(zh->state[sid] == ZOO_CONNECTING_STATE) {
            *interest |= ZOOKEEPER_WRITE;
        }

    }
    return api_epilog(zh,ZOK);
}

static int check_events(zhandle_t *zh, int events, int sid)
{
    if (zh->fd[sid] == -1)
        return ZINVALIDSTATE;
    if ((events&ZOOKEEPER_WRITE)&&(zh->state[sid] == ZOO_CONNECTING_STATE)) {
        int rc, error;
        socklen_t len = sizeof(error);
        rc = getsockopt(zh->fd[sid], SOL_SOCKET, SO_ERROR, &error, &len);
        /* the description in section 16.4 "Non-blocking connect"
         * in UNIX Network Programming vol 1, 3rd edition, points out
         * that sometimes the error is in errno and sometimes in error */
        if (rc < 0 || error) {
            if (rc == 0)
                errno = error;
	    //printf("set %d to -1\n", sid);
	    zh->fd[sid] = -1;
            return handle_socket_error_msg(zh, __LINE__,ZCONNECTIONLOSS, sid,
                "server refused to accept the client");
        }
        if((rc=prime_connection(zh, sid))!=0)
            return rc;
        LOG_INFO(("initiated connection to server [%s]",
                format_endpoint_info(&zh->addrs[sid])));
        return ZOK;
    }
    if (zh->to_send[sid].head && (events&ZOOKEEPER_WRITE)) {
        /* make the flush call non-blocking by specifying a 0 timeout */
        int rc=flush_send_queue(zh,0,sid);
        if (rc < 0)
            return handle_socket_error_msg(zh,__LINE__,ZCONNECTIONLOSS, sid,
                "failed while flushing send queue");
    }
    if (events&ZOOKEEPER_READ) {
        int rc;
        if (zh->input_buffer[sid] == 0) {
            zh->input_buffer[sid] = allocate_buffer(0,0);
        }

        rc = recv_buffer(zh->fd[sid], zh->input_buffer[sid]);
	//printf("check read event %d\n", rc);
        if (rc < 0) {
            return handle_socket_error_msg(zh, __LINE__,ZCONNECTIONLOSS, sid, 
                "failed while receiving a server response");
        }
        if (rc > 0) {
            gettimeofday(&zh->last_recv[sid], 0);
            if (zh->input_buffer[sid] != &zh->primer_buffer[sid]) {
		//printf("queue buffer\n");
                queue_buffer(&zh->to_process[sid], zh->input_buffer[sid], 0);
            } else  {
                int64_t oldid,newid;
		//printf("get prime response\n");
                //deserialize
                deserialize_prime_response(&zh->primer_storage[sid], zh->primer_buffer[sid].buffer);
                /* We are processing the primer_buffer, so we need to finish
                 * the connection handshake */
                oldid = zh->client_id.client_id;
                newid = zh->primer_storage[sid].sessionId;
		//printf("oldid = %lld, newid = %lld\n", oldid, newid);
                if (oldid != 0 && oldid != newid) {
                    zh->state[sid] = ZOO_EXPIRED_SESSION_STATE;
                    errno = ESTALE;
                    return handle_socket_error_msg(zh,__LINE__,ZSESSIONEXPIRED, sid,
                            "sessionId=%#llx has expired.",oldid);
                } else {
                    zh->recv_timeout = zh->primer_storage[sid].timeOut;
		    init_xid(zh->storage, zh->primer_storage[sid].cxid);
                    //zh->client_id.client_id = newid;
                 
                    zh->state[sid] = ZOO_CONNECTED_STATE;
		    
                    /*printf("session %d establishment complete on server [%s], sessionId=%#llx, negotiated timeout=%d, cxid=%d\n",
                              sid, format_endpoint_info(&zh->addrs[sid]),
                              newid, zh->recv_timeout, zh->primer_storage[sid].cxid);*/
                    zh->input_buffer[sid] = 0; // just in case the watcher calls zookeeper_process() again
                }
            }
            zh->input_buffer[sid] = 0;
        } else {
            // zookeeper_process was called but there was nothing to read
            // from the socket
            return ZNOTHING;
        }
    }
    return ZOK;
}

void api_prolog(zhandle_t* zh)
{
    inc_ref_counter(zh,1);
}

int api_epilog(zhandle_t *zh,int rc)
{
    if(inc_ref_counter(zh,-1)==0 && zh->close_requested!=0)
	;
        //zookeeper_close(zh);
    return rc;
}




static void isSocketReadable(zhandle_t* zh, int sid)
{
    struct pollfd fds;
    fds.fd = zh->fd[sid];
    fds.events = POLLIN;
    if (poll(&fds,1,0)<=0) {
        // socket not readable -- no more responses to process
        zh->socket_readable[sid].tv_sec=zh->socket_readable[sid].tv_usec=0;
    }else{
        gettimeofday(&zh->socket_readable[sid],0);
    }
}

static void checkResponseLatency(zhandle_t* zh, int sid)
{
    int delay;
    struct timeval now;

    if(zh->socket_readable[sid].tv_sec==0)
        return;

    gettimeofday(&now,0);
    delay=calculate_interval(&zh->socket_readable[sid], &now);
    if(delay>20)
        LOG_DEBUG(("The following server response has spent at least %dms sitting in the client socket recv buffer",delay));

    zh->socket_readable[sid].tv_sec=zh->socket_readable[sid].tv_usec=0;
}

int zookeeper_process(zhandle_t *zh, int events, int sid)
{
    buffer_list_t *bptr;
    int rc;
    //printf("zookeeper process %d %d\n", sid, zh);
    if (zh==NULL)
        return ZBADARGUMENTS;
    //printf("zookeeper process %d %d\n", sid, zh);
    if (is_unrecoverable(zh))
        return ZINVALIDSTATE;
    api_prolog(zh);
    IF_DEBUG(checkResponseLatency(zh, sid));
    rc = check_events(zh, events, sid);
    if (rc!=ZOK)
        return api_epilog(zh, rc);

    IF_DEBUG(isSocketReadable(zh, sid));

    while (rc >= 0 && (bptr=dequeue_buffer(&zh->to_process[sid]))) {
        struct ReplyHeader hdr;
        struct iarchive *ia = create_buffer_iarchive(
                                    bptr->buffer, bptr->curr_offset);
        deserialize_ReplyHeader(ia, "hdr", &hdr);
        if (hdr.zxid > 0) {
	    enter_critical(zh);
            zh->last_zxid = hdr.zxid;
	    leave_critical(zh);
        } else {
            // fprintf(stderr, "Got %#x for %#x\n", hdr.zxid, hdr.xid);
        }

	handle_response(zh->storage, &hdr, ia, sid);
	free_buffer(bptr);

        close_buffer_iarchive(&ia);

    }
    return api_epilog(zh,ZOK);}

int zoo_state(zhandle_t *zh, int sid)
{
    if(zh!=0)
        return zh->state[sid];
    return 0;
}




int zookeeper_close(zhandle_t *zh, int xid)
{
    int rc=ZOK;
    int i;
    if (zh==0)
        return ZBADARGUMENTS;

    zh->close_requested=1;
    if (inc_ref_counter(zh,0)!=0) {

        adaptor_finish(zh);
        return ZOK;
    }
    for(i=0; i<SERVER_COUNT; i++){
    	if(zh->state[i]==ZOO_CONNECTED_STATE){
            struct oarchive *oa;
            struct RequestHeader h = { .xid = xid, .type = CLOSE_OP};
            LOG_INFO(("Closing zookeeper sessionId=%#llx to [%s]\n",
                zh->client_id.client_id,format_current_endpoint_info(zh, i)));
            oa = create_buffer_oarchive();
            rc = serialize_RequestHeader(oa, "header", &h);
            rc = rc < 0 ? rc : queue_buffer_bytes(&zh->to_send[i], get_buffer(oa),
                get_buffer_len(oa));
            /* We queued the buffer, so don't free it */
            close_buffer_oarchive(&oa, 0);
            if (rc < 0) {
            	rc = ZMARSHALLINGERROR;
            	goto finish;
            }

            /* make sure the close request is sent; we set timeout to an arbitrary
             * (but reasonable) number of milliseconds since we want the call to block*/
            rc=adaptor_send_queue(zh, 3000, i);
        }
    }

finish:
    destroy(zh);
    adaptor_destroy(zh);
    free(zh);
    return rc;
}




/* specify timeout of 0 to make the function non-blocking */
/* timeout is in milliseconds */
int flush_send_queue(zhandle_t*zh, int timeout, int sid)
{
    int rc= ZOK;
    struct timeval started;
    gettimeofday(&started,0);
    // we can't use dequeue_buffer() here because if (non-blocking) send_buffer()
    // returns EWOULDBLOCK we'd have to put the buffer back on the queue.
    // we use a recursive lock instead and only dequeue the buffer if a send was
    // successful
    lock_buffer_list(&zh->to_send[sid]);
    while (zh->to_send[sid].head != 0&& zh->state[sid] == ZOO_CONNECTED_STATE) {
        if(timeout!=0){
            int elapsed;
            struct pollfd fds;
            struct timeval now;
            gettimeofday(&now,0);
            elapsed=calculate_interval(&started,&now);
            if (elapsed>timeout) {
                rc = ZOPERATIONTIMEOUT;
                break;
            }
            fds.fd = zh->fd[sid];
            fds.events = POLLOUT;
            fds.revents = 0;
            rc = poll(&fds, 1, timeout-elapsed);
            if (rc<=0) {
                /* timed out or an error or POLLERR */
                rc = rc==0 ? ZOPERATIONTIMEOUT : ZSYSTEMERROR;
                break;
            }
        }

        rc = send_buffer(zh->fd[sid], zh->to_send[sid].head);
        if(rc==0 && timeout==0){
            /* send_buffer would block while sending this buffer */
            rc = ZOK;
            break;
        }
        if (rc < 0) {
            rc = ZCONNECTIONLOSS;
            break;
        }
        // if the buffer has been sent successfully, remove it from the queue
        if (rc > 0)
            remove_buffer(&zh->to_send[sid]);
        gettimeofday(&zh->last_send[sid], 0);
        rc = ZOK;
    }
    unlock_buffer_list(&zh->to_send[sid]);
    return rc;
}

const char* zerror(int c)
{
    switch (c){
    case ZOK:
      return "ok";
    case ZSYSTEMERROR:
      return "system error";
    case ZRUNTIMEINCONSISTENCY:
      return "run time inconsistency";
    case ZDATAINCONSISTENCY:
      return "data inconsistency";
    case ZCONNECTIONLOSS:
      return "connection loss";
    case ZMARSHALLINGERROR:
      return "marshalling error";
    case ZUNIMPLEMENTED:
      return "unimplemented";
    case ZOPERATIONTIMEOUT:
      return "operation timeout";
    case ZBADARGUMENTS:
      return "bad arguments";
    case ZINVALIDSTATE:
      return "invalid zhandle state";
    case ZAPIERROR:
      return "api error";
    case ZNONODE:
      return "no node";
    case ZNOAUTH:
      return "not authenticated";
    case ZBADVERSION:
      return "bad version";
    case  ZNOCHILDRENFOREPHEMERALS:
      return "no children for ephemerals";
    case ZNODEEXISTS:
      return "node exists";
    case ZNOTEMPTY:
      return "not empty";
    case ZSESSIONEXPIRED:
      return "session expired";
    case ZINVALIDCALLBACK:
      return "invalid callback";
    case ZINVALIDACL:
      return "invalid acl";
    case ZAUTHFAILED:
      return "authentication failed";
    case ZCLOSING:
      return "zookeeper is closing";
    case ZNOTHING:
      return "(not error) no server responses to process";
    case ZSESSIONMOVED:
      return "session moved to another server, so operation is ignored";
    }
    if (c > 0) {
      return strerror(c);
    }
    return "unknown error";
}


static const char* format_endpoint_info(const struct sockaddr_storage* ep)
{
    static char buf[128];
    char addrstr[128];
    void *inaddr;
    int port;
    if(ep==0)
        return "null";

#if defined(AF_INET6)
    if(ep->ss_family==AF_INET6){
        inaddr=&((struct sockaddr_in6*)ep)->sin6_addr;
        port=((struct sockaddr_in6*)ep)->sin6_port;
    } else {
#endif
        inaddr=&((struct sockaddr_in*)ep)->sin_addr;
        port=((struct sockaddr_in*)ep)->sin_port;
#if defined(AF_INET6)
    }
#endif
    inet_ntop(ep->ss_family,inaddr,addrstr,sizeof(addrstr)-1);
    sprintf(buf,"%s:%d",addrstr,ntohs(port));
    return buf;
}

static const char* format_current_endpoint_info(zhandle_t* zh, int sid)
{
    return format_endpoint_info(&zh->addrs[sid]);
}

void zoo_deterministic_conn_order(int yesOrNo)
{
    disable_conn_permute=yesOrNo;
}
